package com.atguigu.day04;

import com.atguigu.utils.IntSource;
import com.atguigu.utils.IntStatistic;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;

public class Example3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new IntSource())
                .keyBy(new KeySelector<Integer, String>() {
                    @Override
                    public String getKey(Integer in) throws Exception {
                        if (in % 2 == 0) {
                            return "even";
                        } else {
                            return "odd";
                        }
                    }
                })
                .process(new KeyedProcessFunction<String, Integer, String>() {
                    private ValueState<IntStatistic> acc;
                    private ValueState<Boolean> flag;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        acc = getRuntimeContext().getState(
                                new ValueStateDescriptor<IntStatistic>(
                                        "acc",
                                        Types.POJO(IntStatistic.class)
                                )
                        );
                        flag = getRuntimeContext().getState(
                                new ValueStateDescriptor<Boolean>(
                                        "flag",
                                        Types.BOOLEAN
                                )
                        );
                    }

                    @Override
                    public void processElement(Integer in, Context ctx, Collector<String> out) throws Exception {
                        if (acc.value() == null) {
                            acc.update(new IntStatistic(
                                    in, in, in, 1, in
                            ));
                        } else {
                            IntStatistic oldAcc = acc.value();
                            IntStatistic newAcc = new IntStatistic(
                                    Math.min(in, oldAcc.min),
                                    Math.max(in, oldAcc.max),
                                    in + oldAcc.sum,
                                    1 + oldAcc.count,
                                    (in + oldAcc.sum) / (1 + oldAcc.count)
                            );
                            acc.update(newAcc);
                        }

                        if (flag.value() == null) {
                            long currTs = ctx.timerService().currentProcessingTime();
                            ctx.timerService().registerProcessingTimeTimer(currTs + 10 * 1000L);
                            flag.update(true);
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        out.collect(ctx.getCurrentKey() + ": " + acc.value());
                        flag.clear();
                    }
                })
                .print();

        env.execute();
    }
}
